package org.sg20.graphbuilder;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.ArrayList;


public class gx2 {
    public static class map extends Mapper<LongWritable, Text, Text, Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split("\t");
            String[] split1 = split[0].split(",");
            Text text = new Text(split1[0].substring(1));
//            System.out.println(text.toString());
            Text text1 = new Text(split1[1].substring(0,split1[1].length()-1)+"\t"+split[1]);
//            System.out.println(text1);
            context.write(text,text1);
        }
    }
    private static class reduce extends Reducer<Text,Text,Text,Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            ArrayList<String> arrayList = new ArrayList<>();
            int sum=0;
            for (Text i : values) {
                String[] split = i.toString().split("\t");
                sum+=Integer.parseInt(split[1]);
                arrayList.add(i.toString());
            }
            StringBuilder sb=new StringBuilder("[");
            String destName;
            double weight;
            String[] split = arrayList.get(0).split("\t");
            destName=split[0];
            weight=Double.parseDouble(split[1])/sum;
            sb.append(destName).append(',').append(String.format("%.3f",weight));
            for(int i=1;i<arrayList.size();i++){
                split = arrayList.get(i).split("\t");
                destName=split[0];
                weight=Double.parseDouble(split[1])/sum;
                sb.append('|').append(destName).append(',').append(String.format("%.3f",weight));
            }
            sb.append(']');
            Text text = new Text(sb.toString());
            context.write(key,text);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();

        //Parse arguments
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Please input: xxx.jar gx2 [InputPath] [OutputPath].");
            System.exit(1);
        }

        Job job = Job.getInstance(conf, "gx2");
        job.setJarByClass(gx2.class);

        job.setMapperClass(gx2.map.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(gx2.reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);


        String in = otherArgs[0];
        String out = otherArgs[1]+"/Build";



        FileInputFormat.setInputPaths(job, new Path(in));
        FileOutputFormat.setOutputPath(job, new Path(out));
        FileSystem fs = FileSystem.get(new Configuration());
        String outPath = out;
        if (fs.exists(new Path(outPath))){
            fs.delete(new Path(outPath),true);
        }
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

